package com.zyh.flink.day06.function;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

//以word-count为例，演示processFunction
public class ProcessFunctionJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStreamSource = environment.socketTextStream("hadoop10", 9999);

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split("\\s+");

                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        }).keyBy(t -> t.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = windowedStream.process(new MyProcessFunction());
        
        result.print();
        
        environment.execute("processFunction");
    }
}
/*
    4个类型参数：
    In 输入的元素类型 Tuple2<String,Integer>
    OUT 输出结果类型 Tuple2<String,Integer>
    KEY keyedStream分组依据类型
    W extends Window 窗口类型
 */

class MyProcessFunction extends ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow> {

    /**
     *
     * @param key 流上一组的分组依据值
     * @param context 上下文对象，可以通过这个对象获取到元数据信息
     * @param elements 窗口中所有元素：
     * @param out collector，用来把数据发送给下游
     */
    @Override
    public void process(String key, ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>.Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
        System.out.println("elements = " + elements);
        // 获取元数据信息
        TimeWindow window = context.window();
        long start = window.getStart();
        long end = window.getEnd();
        //每一个窗口都是左闭右开的区间
        System.out.println("start="+start+",end="+end);

        //统计单词个数
        int count = 0;
        for (Tuple2<String, Integer> element : elements) {
            count += element.f1;
        }

        out.collect(Tuple2.of(key,count));
    }
}